Fixes #39446.

The bug

An unbounded source SDF that returns ProcessContinuation.resume() has its residual re-queued by reElementResiduals as a pending element carrying the input element's event time, which for an Impulse rooted source is mtime.MinTimestamp. That pins minPendingTimestampLocked, so the stage's watermark never advances.

updateWatermarks returns no refreshes when the output watermark does not move, and PersistBundle marked only the producing stage as changed. So a consumer that had just been handed pending elements was never inserted into em.changedStages, bundleReady was never called for it, and its elements accumulated forever.

This is worse than the issue describes: it starves plain ordinary consumers too, not only stateful ones, and it survives transitively through a chain of stages. checkForQuiescence cannot catch it, because the source stage stays schedulable, so the job live locks rather than failing fast.

Reproduced at the element manager level before fixing. Pre-fix, across 5 source bundles:

round 0: src out=-inf | sink pending=1 scheduled=0
round 4: src out=-inf | sink pending=5 scheduled=0

The fix

1. PersistBundle records the consumers that accepted data into stageState.consumersWithNewData, and updateWatermarks hands that set to the scheduler on the newOut <= ss.output path, clearing it on the advancing path where the normal return already lists every consumer. Routing the wake-up through updateWatermarks is what keeps a pipeline whose watermark advances on exactly its previous schedule.
2. The recording is gated on em.sawResidual, latched the first time any bundle returns a residual. A pipeline that never self checkpoints never populates the set, so its bundle scheduling is unchanged. This matters: an earlier revision that recorded unconditionally changed bundle packing and broke TestRunner_Pipelines/flatten_to_sideInput at -count=20.
3. bundleReady lets a stateful stage with no side inputs run on pending data alone, since statefulStageKind.buildEventTimeBundle already takes data at any watermark and gates only timers. Its stillSchedulable now requires buildable work so a key holding only a future timer cannot spin against the relaxed gate, and such a key no longer consumes the OneKeyPerBundle slot, is not marked in progress, and does not hold the bundle's minimum timestamp.

What is deliberately unchanged

• The watermark value. BundleApplication.output_watermarks says an unreported estimate means MIN_TIMESTAMP, and Java's ProcessFn holds at futureOutputWatermark ?? elementTimestamp. A pinned MIN_TIMESTAMP output watermark for a source that reports no estimate is the spec, so the TODO the issue points at is not where the defect is.
• Aggregate stages stay on the watermark gate. A GBK genuinely needs window closure or a trigger.
• Any stage that reads a side input stays on the watermark gate, so such a consumer still waits under a residual pinned source. Side input readiness is derived from the main watermark, and GetSideData filters non global windows by the bundle watermark, so relaxing this would let a bundle read a partial or empty side input. That is worse than waiting. Fixing it properly means changing how side inputs are materialized for every stage, which I would rather do separately with your input.

Testing

New elementmanager_continuation_test.go, all teeth checked by reverting each half of the fix independently:

• TestPersistBundle_ContinuationResidualConsumers, ordinary and stateful consumers of a source that self checkpoints forever.
• TestPersistBundle_ContinuationResidualTransitive, src self checkpoints, mid returns no residual of its own, sink two stages down.
• TestPersistBundle_ContinuationResidualWatermark, pins the output_watermarks contract in both directions.
• TestStatefulBuildEventTimeBundle_OneKeyPerBundle, a key headed by a future timer must not consume the single key slot, and must not enter newKeys or hold minTs.

Green: go test ./pkg/beam/runners/prism/... -count=1, -run "TestRunner_Pipelines/flatten" -count=20, and the new tests at -count=50.

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Choose reviewer(s) and mention them in a comment (R: @username).
• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md